package com.max.rabbitmq.learn.retry;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.max.rabbitmq.learn.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @description 简单模式接收者
 * @author yandong
 * @create 2021年4月06日 12:15:50
 */
public class Recv {
    private static final String NORMAL = "max.normal";
    private static final String RETRY = "max.retry";
    private static final String FAIL = "max.fail";
    private static final Integer RETRY_INTERVAL = 5;

    private static final String EXCHANGE_NORMAL = "exchange.max.normal";
    private static final String EXCHANGE_RETRY = "exchange.max.retry";
    private static final String EXCHANGE_FAIL = "exchange.max.fail";

    public static void main(String[] args) throws IOException, TimeoutException {

        //获取连接
        Connection connection = ConnectionUtil.getConnection();

        //获取chanel chanel是操作rabbitmq broker的接口，即操作amqp协议的对象
        Channel channel = connection.createChannel();

        //正常队列参数
        Map normalArgs = new HashMap();
        //重试队列参数
        Map retryArgs = new HashMap();
        //死信队列参数
        Map failArgs = new HashMap();

        normalArgs.put("x-dead-letter-exchange", EXCHANGE_FAIL);

        retryArgs.put("x-dead-letter-exchange", EXCHANGE_NORMAL);
        retryArgs.put("x-message-ttl", 60 * 1000);

        channel.exchangeDeclare(EXCHANGE_NORMAL, BuiltinExchangeType.TOPIC, true);
        channel.exchangeDeclare(EXCHANGE_RETRY, BuiltinExchangeType.TOPIC, true);
        channel.exchangeDeclare(EXCHANGE_FAIL, BuiltinExchangeType.TOPIC, true);

        //正常队列
        channel.queueDeclare(NORMAL, true, false, false, normalArgs);
        //重试队列
        channel.queueDeclare(RETRY, true, false, false, retryArgs);
        //死信队列
        channel.queueDeclare(FAIL, true, false, false, failArgs);

        channel.queueBind(NORMAL, EXCHANGE_NORMAL, "#");
        channel.queueBind(RETRY, EXCHANGE_RETRY, "#");
        channel.queueBind(FAIL, EXCHANGE_FAIL, "#");

        channel.basicConsume(NORMAL, false, ((consumerTag, message) -> {
            System.out.println("/****************************/");
            System.out.println("recv from exchange = " + message.getEnvelope().getExchange());
            System.out.println("routingKey = " + message.getEnvelope().getRoutingKey());
            System.out.println("messageId = " + message.getProperties().getMessageId());
            System.out.println("message = " + new String(message.getBody(), "UTF-8"));
            Integer retryCount = getRetryCount(message.getProperties());
            System.out.println("第" + (retryCount + 1) + "次重试");
            //重试超过3次之后直接扔到死信队列
            if (retryCount >= 2) {
                channel.basicNack(message.getEnvelope().getDeliveryTag(), false, false);
            } else {
                //计算重新投递间隔
                Integer interval = RETRY_INTERVAL * 1000;
                AMQP.BasicProperties oldProps = message.getProperties();
                //这一步是必须的 否则之后的x-death的count不会增加
                AMQP.BasicProperties newProps = new AMQP.BasicProperties().builder()
                        .headers(oldProps.getHeaders())
                        .expiration(interval + "")
                        .messageId(oldProps.getMessageId())
                        .build();
                channel.basicPublish(EXCHANGE_RETRY, message.getEnvelope().getRoutingKey(), newProps, message.getBody());
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        }), (consumerTag -> {
        }));

    }

    /**
     * 获得重试次数
     *
     * @param properties message属性
     * @return Integer 重试次数
     */
    private static Integer getRetryCount(AMQP.BasicProperties properties) {
        Map header = properties.getHeaders();
        if (header != null) {
            String jsonArrayString = JSONArray.toJSONString(header.get("x-death"));
            List list = JSONObject.parseObject(jsonArrayString, List.class);
            Map deathInfo = JSONObject.parseObject(JSONObject.toJSONString(list.get(0)), Map.class);
            return Integer.valueOf(deathInfo.get("count").toString());
        }
        return 0;
    }
}
